import IAsgard from "./IAsgard";
import {BehaviorSubject, Observable, ReplaySubject, Subject, Subscriber} from "rxjs";
import {AsgardState, Notify} from "./DAsgard";
import {OnTerminalSubscriber, Payload, RSocket, RSocketConnector} from "rsocket-core";
import {WebsocketClientTransport} from "rsocket-websocket-client";
import {encodeCompositeMetadata, encodeRoute, WellKnownMimeType} from 'rsocket-composite-metadata'
import {Cancellable, OnExtensionSubscriber, OnNextSubscriber} from "rsocket-core/dist/RSocket";
import APPLICATION_JSON = WellKnownMimeType.APPLICATION_JSON;
import MESSAGE_RSOCKET_COMPOSITE_METADATA = WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA;
import MESSAGE_RSOCKET_ROUTING = WellKnownMimeType.MESSAGE_RSOCKET_ROUTING;

export default class Asgard implements IAsgard {

    public readonly notifySubject: ReplaySubject<Notify<any>>;
    public readonly stateSubject: BehaviorSubject<AsgardState>;
    public readonly errorSubject: Subject<Error|undefined>;

    private connector!: RSocketConnector;
    private token?: string;
    private clientId?: string;
    private client?: RSocket;
    private readonly resumeTime: number = 1000 * 60; // 恢复时间为1分钟
    private resumeStart: number = 0; // 恢复开始时间
    public resumeAttempt: number = 0
    private _status: AsgardState = AsgardState.READY;
    private url: string
    get status(): AsgardState {
        return this._status;
    }

    set status(value: AsgardState) {
        this._status = value;
        this.stateSubject.next(value);
    }

    constructor(url: string, token?: string, clientId?: string) {
        this.url = url
        this.token = token
        this.clientId = clientId
        this.notifySubject = new ReplaySubject<Notify<any>>()
        this.stateSubject = new BehaviorSubject<AsgardState>(AsgardState.READY)
        this.errorSubject = new Subject<Error|undefined>()
    }
    connect(): Promise<RSocket> {
        this.isManualDisConnect = false
        if (!this.connector) {
            this.connector = new RSocketConnector({
                setup: {
                    dataMimeType: APPLICATION_JSON.string,
                    metadataMimeType: MESSAGE_RSOCKET_COMPOSITE_METADATA.string,
                    keepAlive: 3000, // 每3s发送一次心跳
                    lifetime: 10000, // 10s内没有收到心跳则认为连接断开
                    payload: {
                        data: this.setupData(),
                    },
                },
                fragmentation: {
                    maxOutboundFragmentSize: 1024 * 1024 * 2,
                },
                transport: new WebsocketClientTransport({
                    url:this.url,
                    wsCreator: (url: string | URL) => {
                        const ws = new WebSocket(url)
                        ws.onmessage = (event: MessageEvent) => {
                            const dataArray = Buffer.from(event.data)
                            if (dataArray[4] === 0x38&&dataArray.length===14) {
                                this.resumeStart = 0
                                this.resumeAttempt = 0
                                this.status = AsgardState.CONNECTED
                                console.warn('resume success')
                            } else if (dataArray[4] === 0x2c&&dataArray.length===14) {
                                this.status = AsgardState.RECOVERYFAILED
                                this.resumeStart = 0
                                this.client?.close()
                                this.client = undefined
                                this.resumeAttempt = 0
                                console.warn('resume failed')
                            }
                        }
                        return ws
                    },
                    debug: false,
                }),
                responder: {
                    requestResponse: (payload: Payload, responderStream: OnTerminalSubscriber & OnNextSubscriber & OnExtensionSubscriber)=>{
                        const data: Notify<any> = payload.data ? JSON.parse(payload.data.toString()) : {}
                        if (data["@type"].toLowerCase() === "notify") {
                          this.notifySubject.next(data)
                        }
                        return {
                          cancel() {
                            console.warn("cancel");
                          },
                          onExtension(extendedType: number, content: Buffer | null | undefined, canBeIgnored: boolean){
                            console.log('onExtension',extendedType, content, canBeIgnored)
                          }
                        } as Cancellable&OnExtensionSubscriber;
                    },
                    fireAndForget: (payload: Payload, responderStream: OnTerminalSubscriber) => {
                        const data: Notify<any> = payload.data ? JSON.parse(payload.data.toString()) : {}
                        if (data['@type'].toLowerCase() === 'notify') {
                            this.notifySubject.next(data)
                        }
                        responderStream.onComplete()
                        return {
                            cancel() {
                                console.log('cancel')
                            }
                        };
                    },
                },
                // lease: {
                //     maxPendingRequests: 1000
                // },
                resume: {
                    cacheSize: 1000,
                    tokenGenerator: () => Buffer.from(`-${Date.now()}${this.clientId ? this.clientId : ''}`),
                    reconnectFunction: (attempt: number) => {
                        if (this.isManualDisConnect) {
                            return Promise.reject(new Error('Is Manual DisConnect'))
                        }
                        this.resumeAttempt = attempt
                        console.log('resume attempt:' + attempt)
                        return new Promise<void>((resolve, reject) => {
                            if ((this.resumeStart === 0 || Date.now() - this.resumeStart <= this.resumeTime) && this.status !== AsgardState.RECOVERYFAILED) {
                                this.resumeStart === 0 && (this.resumeStart = Date.now())
                                this.status = AsgardState.RECOVERING
                                setTimeout(resolve, 500 * attempt)
                            } else {
                                this.status = AsgardState.RECOVERYFAILED
                                this.resumeStart = 0
                                this.client?.close()
                                this.client = undefined
                                reject(new Error('resume timeout'))
                            }
                        })
                    }
                }
            })
        }
        if (this.status < 4) {
            return new Promise((resolve, reject) => {
                this.status = AsgardState.CONNECTING
                this.connector.connect().then((socket: RSocket) => {
                    this.client = socket
                    socket.onClose((err?: Error) => {
                        this.status = AsgardState.DISCONNECT
                        if (this.isManualDisConnect) {
                            console.log('Asgard manual disconnect');
                        } else {
                            console.warn('Asgard is disconnected:' + err?.message)
                            this.onError(err)
                        }
                    })
                    this.status = AsgardState.CONNECTED
                    resolve(socket)
                }).catch((err: Error) => {
                    console.warn('Asgard is connection error:' + err?.message)
                    this.status = AsgardState.CONNECTIONERROR
                    reject(err)
                })
            })
        } else {
            console.warn('Asgard is recovering, please wait for a moment')
            return Promise.resolve(this.client!)
        }
    }

    private isManualDisConnect = false
    disconnect(): void {
        this.isManualDisConnect = true
        this.client?.close()
    }

    onError(err: Error|undefined): void {
        this.errorSubject.next(err)
    }

    broadcast<Q, S>(path: string, data?: Q, initialRequestN: number=2147483647): Subject<S> {
        const subject = new Subject<S>()
        const requestChannel = this.client?.requestChannel(
            this.createRequestPayload(path, data, new Map<string, string>([['method', 'broadcast']])),
            initialRequestN,
            false,
            {
                // onExtension(extendedType: number, content: Buffer | null | undefined, canBeIgnored: boolean): void {
                // }, request(requestN: number): void {
                // },
                onNext: (payload: Payload) => {
                    try {
                        subject.next(JSON.parse(payload.data.toString()))
                    } catch (e: any) {
                        // 消息处理异常
                        subject.next(payload.data.toString())
                    }
                },
                onComplete: subject.complete,
                onError: (err: Error) => {
                    this.onError(err)
                    subject.error(err)
                },
                cancel: subject.unsubscribe
            }
        )
        subject.subscribe(null,null,()=>{
            requestChannel?.cancel()
        })
        return subject;
    }

    get<Q, B>(path: string, data?: Q): Promise<B> {
        return new Promise<B>((resolve, reject) => {
            this.client?.requestResponse(this.createRequestPayload(path, data, new Map<string, string>([['method', 'get']])), {
                onComplete: resolve,
                onError: (err: Error) => {
                    this.onError(err)
                    reject(err)
                },
                onNext: (payload: Payload) => {
                    try {
                        resolve(payload.data?JSON.parse(payload.data.toString()):'')
                    } catch (e: any) {
                        // // 消息处理异常
                        // this.onError(e)
                        resolve(payload.data?.toString())
                    }
                },
            })
        })
    }

    notify<S extends { route: string, body: string }>(): Observable<Notify<S>> {
        return this.notifySubject;
    }

    post<Q>(path: string, data?: Q): Promise<void> {
        return new Promise<void>((resolve, reject) => {
            this.client?.requestResponse(this.createRequestPayload(path, data, new Map<string, string>([['method', 'post']])), {
                onComplete: resolve,
                onError: (err: Error) => {
                    this.onError(err)
                    reject(err)
                },
                onNext: resolve,
            })
        })
    }


    auth(token?: string): Promise<void> {
        if(token)this.token = token
        return this.post('/internal/auth', {token:this.token});
    }

    setClientId(clientId: string): void {
        this.clientId = clientId
    }

    unAuth(): Promise<void> {
        this.token = ''
        return this.post('/internal/unAuth');
    }

    private setupData(): Buffer {
        const data: {
            token?: string
            clientId?: string
        } = {}
        this.token && (data.token = this.token)
        this.clientId && (data.clientId = this.clientId)
        return Buffer.from(
            JSON.stringify(Object.assign(data, {
                clientType: 'browser',
                ignoreAuthFailed: true,
            })),
        )
    }

    private createRequestPayload(
        uri: string,
        data?: any,
        header: Map<string, string> = new Map<string, string>(),
    ): Payload {
        const route: [WellKnownMimeType, Buffer] = [
            MESSAGE_RSOCKET_ROUTING,
            encodeRoute(uri),
        ]
        if (!header.get('sessionId') && this.clientId) header.set('sessionId', this.clientId)
        const obj: any = {}
        for (const key of header.keys())
            obj[key] = header.get(key)

        const json: [WellKnownMimeType, Buffer] = [
            APPLICATION_JSON,
            Buffer.from(JSON.stringify(obj)),
        ]
        let dataBuffer
        if (data) {
            dataBuffer = Buffer.from(
                typeof data == 'object' ? JSON.stringify(data) : data,
            )
        } else {
            dataBuffer = Buffer.from('')
        }
        return {
            data: dataBuffer,
            metadata: encodeCompositeMetadata([route, json]),
        }
    }
}
